package com.chukun.flink.stream.window.process.windows;

import com.chukun.flink.stream.window.source.SourceForWindow;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @author chukun
 * @version 1.0.0
 * @description 窗口操作
 * @createTime 2022年05月22日 23:05:00
 */
public class WindowOperator {

    public static void main(String[] args) throws Exception {

        // 创建运行时环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 添加数据源
        DataStreamSource<Tuple3<String, Integer, String>> streamSource = env.addSource(new SourceForWindow(1000, false));

        // 根据元素中的f0字段作为key对数据源进行分组
        KeyedStream<Tuple3<String, Integer, String>, String> keyedStream = streamSource.keyBy((key) -> key.f0);

        // 基于处理时间的滚动窗口，窗口大小5秒
        WindowedStream<Tuple3<String, Integer, String>, String, TimeWindow> tumblingWindow = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(5)));

        // 基于处理时间的滑动窗口，窗口大小5秒，滑动间隙3秒
        WindowedStream<Tuple3<String, Integer, String>, String, TimeWindow> slidingWindow = keyedStream.window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(3)));

        // 基于会话窗口，窗口大小5秒
        WindowedStream<Tuple3<String, Integer, String>, String, TimeWindow> sessionWindow = keyedStream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(8)));

        // 基于全局的窗口，GlobalWindows.create()为窗口分配全局窗口分配器，CountTrigger.of(3)一旦窗口中的数据为3就会触发窗口函数对窗口的计算
        WindowedStream<Tuple3<String, Integer, String>, String, GlobalWindow> globalWindow = keyedStream.window(GlobalWindows.create()).trigger(CountTrigger.of(3));

        // 对滚动窗口 f1 字段进行求和
        SingleOutputStreamOperator<Tuple3<String, Integer, String>> tumblingWindowSum = tumblingWindow.sum("f1");

        // 对滑动窗口 f1 字段进行求和
        SingleOutputStreamOperator<Tuple3<String, Integer, String>> slidingWindowSum = slidingWindow.sum("f1");

        // 对会话窗口 f1 字段进行求和
        SingleOutputStreamOperator<Tuple3<String, Integer, String>> sessionWindowSum = sessionWindow.sum("f1");

        // 全局窗口计算
        SingleOutputStreamOperator<Tuple3<String, Integer, String>> globalWindowSum = globalWindow.sum("f1");

        // 滚动窗口元素打印
        tumblingWindowSum.print("^@^滚动窗口中元素求和结果: ");

        // 滑动窗口元素打印
        slidingWindowSum.print("^#^滑动窗口中元素求和结果: ");

        // 会话窗口元素打印
        sessionWindowSum.print("^$^会话窗口中元素求和结果: ");

        // 全局窗口计算
        globalWindowSum.print("^%^会话窗口中元素求和结果:");

        env.execute("WindowOperator");
    }
}
